from kafka import KafkaProducer, KafkaConsumer, KafkaClient
from kafka.errors import kafka_errors
import traceback
import json
import logging


class Client:
    def __init__(self, clusters):
        """
        :param clusters:
        """
        self.client = KafkaClient(
            bootstrap_servers=clusters[0],
            api_version=(0, 10)
        )

    def add_topic(self, topic):
        return self.client.add_topic(topic=topic)

    def close(self):
        try:
            self.client.close()
        except:
            pass

    def __del__(self):
        self.close()

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()


class Producer:
    def __init__(self, clusters):
        self.client = KafkaProducer(
            bootstrap_servers=clusters[0],
            api_version=(0, 10)
        )

    def send(self, topic, value):
        return self.client.send(topic=topic, value=value)

    def close(self):
        try:
            self.client.close()
        except:
            pass

    def __del__(self):
        self.close()

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()


class Consumer:
    def __init__(self, bootstrap_servers):
        self.client = KafkaConsumer(bootstrap_servers=bootstrap_servers, enable_auto_commit=True,
                                    auto_offset_reset='earliest', group_id="test")

    def receive(self, topics):
        self.client.subscribe(topics=topics)
        while True:
            msg = self.client.poll(max_records=1)
            if msg:
                yield msg
                self.client.commit()

    def close(self):
        try:
            logging.info("close kafka")
            print("close kafka")
            self.client.close()
        except:
            pass

    def __del__(self):
        self.close()

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()


if __name__ == '__main__':
    # p = Producer(["127.0.0.1:32979"])
    # a = p.send(topic="lijiacai", value=b"test")
    # a = p.send(topic="lijiacai", value=b"test")
    # a = p.send(topic="lijiacai", value=b"test")
    # a = p.send(topic="lijiacai", value=b"test")
    # a = p.send(topic="lijiacai", value=b"test")
    # print(a.get())

    # c = Consumer(bootstrap_servers=["127.0.0.1:32979"])
    # print("======")
    # for i in c.receive(["lijiacai"]):
    #     print(i)
    c = Client(["127.0.0.1:32979"])
    res = c.add_topic("lijiacai13")
    print(res.failed())
    print(res.succeeded())
